package data.connector.rippling.client;

import cn.hutool.core.util.ReUtil;
import com.fasterxml.jackson.databind.JsonNode;
import lombok.*;
import lombok.extern.log4j.Log4j2;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.util.MultiValueMap;
import org.springframework.util.StringUtils;
import org.springframework.web.reactive.function.client.ClientResponse;
import reactor.core.publisher.*;

import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;

/**
 * com.mspbots.sync.connectwis.TimeEntries
 *
 * @author Alex bob(https://github.com/vnobo)
 * @date Created by 2019/6/6
 */
@Log4j2
@Service
@RequiredArgsConstructor
public class ForwardOnlyClient {

    private final BaseClient baseClient;

    FluxProcessor<WaitAndNext, WaitAndNext> processor = DirectProcessor.create();
    FluxSink<WaitAndNext> sink = processor.sink();

    public void waitAndNext(WaitAndNext waitAndNext) {
        this.sink.next(waitAndNext);
    }

    public Flux<JsonNode> processorSubscribe(long delay, MultiValueMap<String, String> params) {
        return this.processor.flatMap(waitAndNext -> Mono.just(waitAndNext)
                .delayElement(Duration.ofMillis(waitAndNext.delay)))
                .flatMap(waitAndNext -> this.fetchClientResponse(waitAndNext.next,params)
                        .flatMapMany(allNoe -> this.manageResponse(this.sink, allNoe, delay)));
    }

    private Flux<JsonNode> manageResponse(FluxSink<WaitAndNext> sink, ClientResponse resp, long delay) {
        if (resp.statusCode() != HttpStatus.OK) {
            sink.error(new IllegalStateException("Status code invalid"));
        }
        WaitAndNext wn = new WaitAndNext();
        wn.delay = delay;
        return resp.bodyToMono(JsonNode.class)
                .flatMapMany(response -> {
                    wn.next = getNextUrl(resp);
                    if (wn.next == null) {
                        return Flux.empty();
                    }
                    sink.next(wn);
                    return Flux.fromIterable(response);
                });
    }

    public Flux<ClientResponse> getRepository(String path, MultiValueMap<String, String> params) {
        log.debug("Forward Only get params:{}.", params);
        return fetchClientResponse(path,params)
                .expand(clientResponse -> {
                    String newUrl = this.getNextUrl(clientResponse);
                    if (newUrl == null) {
                        return Flux.empty();
                    }
                    return this.fetchClientResponse(newUrl,params);
                });
    }

    private Mono<ClientResponse> fetchClientResponse(String url,MultiValueMap<String, String> params) {
        log.debug("Forward Only next url {}", url);
        return this.baseClient.getWebClient().get()
                .uri(uriBuilder -> uriBuilder.path(url).queryParams(params).build())
                .accept(MediaType.APPLICATION_JSON)
                .exchange()
                .onErrorResume(err -> Mono.empty());
    }

    private String getNextUrl(ClientResponse resp) {
        List<String> links = resp.headers().header("link");
        if (links.size() == 0) {
            return null;
        }
        String newUrl = ReUtil.get("<([^<]+?)>; rel=\"next\"", links.get(0), 1);
        if (StringUtils.isEmpty(newUrl)) {
            return null;
        }
        return URLDecoder.decode(newUrl, StandardCharsets.UTF_8);
    }


    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    public static class WaitAndNext {
        private String next;
        private long delay;
    }
}
